package batch;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

public class operator_mappatition {
    public static void main(String[] args) throws Exception {
        //创建运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    //    获取数据
        ArrayList<String> data = new ArrayList<String>();
        data.add("i a bing");
        data.add("i b shui");
        data.add("bing shui");
        DataSet<String> source =  env.fromCollection(data);// data 可以是个list
        //System.out.println(source.getClass().toString());
        System.out.println("*****mapPatition****");
    //    每句话进行处理，得到含有单词的list中，List<String>是返回的List
        source.mapPartition(new MapPartitionFunction<String, String>() {
            public void mapPartition(Iterable<String> iterable, Collector<String> collector) throws Exception {
                //Iterator<String> patitondata  代表分区中的所有元素
                Iterator<String> patitondata = iterable.iterator();
                while (patitondata.hasNext()){
                    String data = patitondata.next();
                    String[] words = data.split(" ");
                    for(String w: words){
                        collector.collect(w);
                    }
                    //System.out.println(patitondata.next());
                    collector.collect("----------");
                }
            }

        }).print();
        System.out.println("*****map patition****");

    }
}
